#/bin/bash

SSH_PORT=22
KAFKA_HOME=

if [[ -f ../config/zk.cfg ]];then
  echo "加载zk配置文件"
  source ../config/zk.cfg
fi

if [[ -f ../config/kafka.cfg ]];then
  echo "加载kafka配置文件"
  source ../config/kafka.cfg
fi

if [[ -f ../config/config.cfg ]];then
  echo "加载config配置文件"
  source ../config/config.cfg
fi

if [[ -n "$1" ]]; then
   KAFKA_TAR_NAME=$1
fi
if [[ -n "$2" ]]; then
   KAFKA_IPS=$2
fi

if [[ -n "$3" ]]; then
   ZK_IPS=$3
fi

if [[ -n "$4" ]]; then
   ZK_PORT=$4
fi

if [[ -n "${workspace}" ]]; then
	KAFKA_HOME=${workspace}/${KAFKA_NAME}
else
  KAFKA_HOME=${KAFKA_NAME}
fi

path=`pwd`
kafkaIps=(${KAFKA_IPS//,/ })
IPS=`ip addr | grep 'state UP' -A2  | grep  "inet"  |awk '{print $2}' | cut -f1 -d'/'`

######## 启动服务 ########

function start(){
  if [[ ! -n "${kafkaIps[0]}" ]]; then
    su ${USER} -c "${KAFKA_HOME}/bin/kafka-stop.sh"
    sleep 3
    su ${USER} -c "${KAFKA_HOME}/bin/kafka-start.sh"
  else
    ssh -p${SSH_PORT} ${kafkaIps[0]} "su ${USER} -c \"${KAFKA_HOME}/bin/kafka-stop.sh\""
    sleep 3
    ssh -p${SSH_PORT} ${kafkaIps[0]} "su ${USER} -c \"source /etc/profile && sh ${KAFKA_HOME}/bin/kafka-start.sh\""
  fi
}

######## 校验 ########
function check(){
	if [[ ! -n ${ZK_IPS} ]]; then
	  echo "ZK_IPS未设置"
	  exit
	fi
	
	if [[ ! -n ${ZK_PORT} ]]; then
	  echo "ZK_PORT未设置"
	  exit
	fi
	
	if [[ ! -n ${KAFKA_NAME} ]]; then
	  echo "KAFKA_NAME未设置"
	  exit
	fi
	
	if [[ ! -n ${KAFKA_HOME} ]]; then
	  echo "KAFKA_HOME未设置"
	  exit
	fi
	
  if [[ ! -f ../data/${KAFKA_TAR_NAME} ]]; then
    echo "未找到kafka安装包"
    exit
  fi
  
  if [[ ! -n "${KAFKA_IPS}" ]]; then
    echo "KAFKA_IPS 未设置"
    exit
  fi
  
  if [[ ! -n "${KAFKA_PORT}" ]]; then
    echo "KAFKA_PORT 未设置"
    exit
  fi
	
  for i in ${!kafkaIps[@]}
  do
    if [[ ! "${IPS[@]}" =~ "${kafkaIps[$i]}" ]] && [[ -n ${kafkaIps[$i]} ]] ;then
      ssh -p${SSH_PORT} ${kafkaIps[$i]} "ls -la" >/dev/null
    fi
  done
}

######## 解压文件 ########

function unzip(){
  
  echo "正在解压kafka安装包。。。。"
  if [[ -d ${KAFKA_HOME} ]]; then
    rm -rf ${KAFKA_HOME}
  fi
  mkdir -p ${KAFKA_HOME}/logs
  tar -zxvf ../data/${KAFKA_TAR_NAME} -C ${KAFKA_HOME} --strip-components 1 >/dev/null
  echo "完成kafka安装包解压。。。。"
}

######## 集群启动脚本生成 ########
function startFile(){
  if [[ ${#kafkaIps[*]} -ge 3 ]]; then
  	echo "#/bin/bash" >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo "KAFKA_IPS=${KAFKA_IPS}" >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo 'kafkaIps=(${KAFKA_IPS//,/ })' >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo 'for i in ${!kafkaIps[@]}' >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo 'do' >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo "ssh -p${SSH_PORT} ${USER}@\${kafkaIps[\$i]} \"${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties \"" >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo 'done' >> ${KAFKA_HOME}/bin/kafka-start.sh
	
	  echo "#/bin/bash" >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo "KAFKA_IPS=${KAFKA_IPS}" >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo 'kafkaIps=(${KAFKA_IPS//,/ })' >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo 'for i in ${!kafkaIps[@]}' >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo 'do' >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo "pids=\`ssh -p ${SSH_PORT} \${kafkaIps[\$i]} \"ps -ef | grep ${KAFKA_HOME} | grep -v grep | grep -v bash\"| awk '{print \$2}'\` " >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo -e "if [[ -n \"\${pids}\" ]]; then\n for j in \${!pids[@]}\n do\n  ssh -p ${SSH_PORT} ${USER}@\${kafkaIps[\$i]} \"kill -9 \${pids[\$j]}\"\n done\nfi" >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo 'done' >> ${KAFKA_HOME}/bin/kafka-stop.sh
  else
    echo "#/bin/bash" >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo "#/bin/bash" >> ${KAFKA_HOME}/bin/kafka-stop.sh
  
    echo "${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties" >> ${KAFKA_HOME}/bin/kafka-start.sh
    echo "pids=\`ps -ef | grep ${KAFKA_HOME} | grep -v grep | grep -v bash | awk '{print \$2}'\`" >> ${KAFKA_HOME}/bin/kafka-stop.sh
    echo -e "if [[ -n \"\${pids}\" ]]; then\n for j in \${!pids[@]}\n do\n  kill -9 \${pids[\$j]}\n done\nfi" >> ${KAFKA_HOME}/bin/kafka-stop.sh
  fi
  chmod +x ${KAFKA_HOME}/bin/kafka-stop.sh
  chmod +x ${KAFKA_HOME}/bin/kafka-start.sh
}

######## 修改配置文件 ########
function cfg(){	
  sed -i "/^zookeeper.connect=/czookeeper.connect=${ZK_IPS//,/:${ZK_PORT},}:${ZK_PORT}"  ${KAFKA_HOME}/config/server.properties
  sed -i "/^log.dirs=/clog.dirs=${KAFKA_HOME}/logs"  ${KAFKA_HOME}/config/server.properties
  echo "port=${KAFKA_PORT}" >> ${KAFKA_HOME}/config/server.properties

  sed -i "/export KAFKA_HEAP_OPTS=/cexport KAFKA_HEAP_OPTS=\"${KAFKA_HEAP_OPTS}\""  ${KAFKA_HOME}/bin/kafka-server-start.sh
}

######## 拷贝安装文件 ########
function send(){
  echo "正在拷贝kafka安装文件。。。"
  j=0
  for i in ${!kafkaIps[@]}
  do
    if [[ ! "${IPS[@]}" =~ "${kafkaIps[$i]}" ]] && [[ -n "${kafkaIps[$i]}" ]] ;then
      ssh -p${SSH_PORT} ${kafkaIps[$i]} "rm -rf ${KAFKA_HOME} && mkdir -p ${KAFKA_HOME}"
      sed -i "/^broker.id=/cbroker.id=${i}" ${KAFKA_HOME}/config/server.properties
      scp -P${SSH_PORT} -r ${KAFKA_HOME} ${kafkaIps[$i]}:${workspace} >> /dev/null
      ssh -p${SSH_PORT} ${kafkaIps[$i]} "chown -R ${USER}:${USER} ${KAFKA_HOME}"
    else
      j=$i
    fi
  done
  sed -i "/^broker.id=/cbroker.id=${j}" ${KAFKA_HOME}/config/server.properties
  chown -R ${USER}:${USER} ${KAFKA_HOME}
  echo "完成拷贝kafka安装文件。。。"
}

set -e
check
unzip
cfg
startFile
send
set +e
start